package com.example.demo.threadpool;


import cn.hutool.core.thread.NamedThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 初始化一个线程池
 *
 * @author cyx
 */
@Slf4j
@Component
public class ThreadPoolManager implements DisposableBean, Serializable {


    private static final long serialVersionUID = -3353544535147496714L;

    private static String DATA_FLINK = "dataxToL1";

    @Autowired
    private SpringBeanAware springBeanAware;

    /**
     * 线程池
     */
    Map<String, ThreadPoolExecutorWrapper> poolMap = new HashMap<>();


    public ThreadPoolExecutorWrapper getDataThreadPool() {
        return poolMap.get(DATA_FLINK);
    }

    @PostConstruct
    private void initPool() {
        initPool(DATA_FLINK);
    }

    private void initPool(String poolName) {
        String poolConfigBeanName = poolName + "ThreadPoolConfig";
        ThreadPoolConfig searchPoolConfig = (ThreadPoolConfig) springBeanAware.getBean(poolConfigBeanName);
        log.info("初始化线程池：名字:{}, 配置：{}", poolName, searchPoolConfig);
        ThreadPoolExecutorWrapper poolWrapper = initPool(poolName, searchPoolConfig);
        poolMap.put(poolName, poolWrapper);
    }

    private ThreadPoolExecutorWrapper initPool(final String configBeanName, ThreadPoolConfig poolConfig) {
        final int minPoolSize = poolConfig.getMinPoolSize();
        final int maxPoolSize = poolConfig.getMaxPoolSize();
        int workQueueSize = poolConfig.getWorkQueueSize();
        Long keepAliveTime = 3000L;
        if (minPoolSize == maxPoolSize) {
            keepAliveTime = 0L;
        }
        //先进先出队列
        ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(workQueueSize);
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(minPoolSize, maxPoolSize, keepAliveTime, TimeUnit.MILLISECONDS,
                queue,
                new NamedThreadFactory(configBeanName, false),
                //当队列满，拒绝新请求时，抛出异常“服务繁忙，处理失败”
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        //记录到业务日志
                        log.warn("线程池{}队列已满,corePoolSize:{},queueSize:{}", configBeanName, minPoolSize, maxPoolSize);
                    }
                });

        ThreadPoolExecutorWrapper wrapper = new ThreadPoolExecutorWrapper();
        wrapper.name = configBeanName;
        wrapper.threadPoolExecutor = threadPoolExecutor;
        wrapper.totalCapacity = poolConfig.getMaxPoolSize() + poolConfig.getWorkQueueSize();
        wrapper.arrayBlockingQueue = queue;
        return wrapper;
    }

    @Override
    public void destroy() {
        for (ThreadPoolExecutorWrapper wrapper : poolMap.values()) {
            wrapper.threadPoolExecutor.shutdown();
        }
    }

    public static class ThreadPoolExecutorWrapper {
        private String name;
        private ThreadPoolExecutor threadPoolExecutor;
        private int totalCapacity;
        private ArrayBlockingQueue<Runnable> arrayBlockingQueue;

        public long getAvailableCapacity() {
            long capacityUsed = getCapacityUsed();
            return totalCapacity - capacityUsed;
        }

        public boolean isShutDown() {
            return threadPoolExecutor.isShutdown() || threadPoolExecutor.isTerminating();
        }

        public <T> Future<T> submit(Callable<T> task) {
            return threadPoolExecutor.submit(task);
        }

        public void submit(Runnable task) {
            threadPoolExecutor.submit(task);
        }

        public String getCapacityInfo() {
            long capacityUsed = getCapacityUsed();
            return String.format("线程池：%s，线程池总容量: %s，已使用容量：%s, 可用容量: %s", name, totalCapacity, capacityUsed, totalCapacity - capacityUsed);
        }

        private int getCapacityUsed() {
            return threadPoolExecutor.getActiveCount() + threadPoolExecutor.getQueue().size();
        }

        public synchronized boolean isFree() {
            return threadPoolExecutor.getQueue().remainingCapacity() > 0;
        }

        /**
         * @param runnable 必须实现equals方法
         * @return
         */
        public boolean contains(Runnable runnable) {
            return arrayBlockingQueue.contains(runnable);
        }
    }

}
